package liquidfs.core;

import java.io.*;
import java.nio.charset.*;
import java.util.*;
import java.util.concurrent.*;

import utility.*;

import com.thoughtworks.xstream.annotations.*;

import liquidfs.*;
import liquidfs.data.*;
import liquidfs.data.ops.*;

@XStreamAlias("operation-queue")
public class OperationQueue {

   private PeerDevice device = null;
   
   @XStreamImplicit()
   private BlockingQueue<BufferedOperation> queue;
   
   @XStreamOmitField
   private RandomAccessFile file = null;
   
   @XStreamOmitField
   private Application app = null;
   
   @XStreamOmitField
   private Serializer serializer = null;
   
   @SuppressWarnings("unchecked")
   public static OperationQueue loadQperationQueue (Application app, PeerDevice device, int capacity, String path) throws IOException {
      
      ByteBufferInputStream header = new ByteBufferInputStream(new String("<list>").getBytes(Charset.forName("UTF-8")));
      ByteBufferInputStream footer = new ByteBufferInputStream(new String("</list>").getBytes(Charset.forName("UTF-8")));
      FileInputStream file = new FileInputStream(path);
      SequenceInputStream in = new SequenceInputStream(new SequenceInputStream(header, file), footer);
      
      Serializer serializer = Serializer.getDefaultSerializer(app);
      List<BufferedOperation> list = (List<BufferedOperation>) serializer.deserialize(in);
      file.close();
      
      OperationQueue queue = new OperationQueue(app, device, capacity, list);
      queue.app = app;
      queue.device = device;
      
      try {
         queue.file = new RandomAccessFile(new File(app.getConfiguration().getQueuePath() + "/" + device.getID() + "_queue.xmls"), "rwd");
         queue.file.seek(queue.file.length());
      } 
      catch (IOException e) {
         throw new UserUncheckedException("Unable to write queue to journal: " + e.getMessage(), e);
      }
      
      return queue;
   }
   
   private OperationQueue (Application app, PeerDevice device, int capacity, List<BufferedOperation> list) {
      this (app, device, capacity);
      
      this.queue.addAll(list);
   }
   
   public OperationQueue (Application app, PeerDevice device, int capacity) {
      this.app = app;
      this.device = device;
      this.queue = new ArrayBlockingQueue<BufferedOperation>(capacity);
      this.serializer = Serializer.getDefaultSerializer(app);
      
      try {
         this.file = new RandomAccessFile(new File(app.getConfiguration().getQueuePath() + "/" + device.getID() + "_queue.xmls"), "rwd");
         this.file.seek(this.file.length());
      } 
      catch (IOException e) {
         throw new UserUncheckedException("Unable to write queue to journal: " + e.getMessage(), e);
      }
   }
   
   public PeerDevice getDevice () {
      return device;
   }
   
   public long getDataSize () {
      long total = 0;
      
      for (BufferedOperation op : queue) {
         total += op.getSize();
      }
      
      return total;
   }
   
   public int getCapacity () {
      return queue.remainingCapacity() + queue.size();
   }
   
   public int size () {
      return queue.size();
   }
   
   // Core queue operations
   public boolean isEmpty () {
      return queue.isEmpty();
   }
   
   public BufferedOperation peek () {
      return queue.peek();
   }

   public void add (BufferedOperation op) {      
      
      try {
         String xml = serializer.serialize(op);
         file.write(xml.getBytes(Charset.forName("UTF-8")));
         file.write("\r\n\r\n".getBytes(Charset.forName("UTF-8")));
      }
      catch (IOException e) {
         throw new UserUncheckedException ("Unable to write to queue journal: " + e.getMessage(), e);
      }
      
      queue.add(op);
   }
   
   public BufferedOperation remove () {
      
      // Remove an operation from the queue
      BufferedOperation removedOperation = queue.remove();
      
      // If that was the last one, clear our saved queue
      if (queue.isEmpty()) {
         try {
            file.setLength(0);
            file.seek(0);
         } 
         catch (IOException e) {
            throw new UserUncheckedException ("Unable to write to queue journal: " + e.getMessage(), e);
         }
         
      }
      
      // Return the removed operation
      return removedOperation;
   }
   
}
